✏️【本日實作紀錄：API 限流防禦與 Resilience 機制】

今天進入微服務的資安與系統韌性強化階段。在生產環境中需要防止 API 被惡意刷單、大量請求造成 Gemini API 額度耗盡或伺服器過載。完成項目包含：

• 導入 Flask-Limiter 模組 — 設定 IP 層級的 API Rate Limiting 機制，針對高昂資源的 /analyze-prescription Endpoint 進行請求頻率限制。
• 自訂 HTTP 429 流量管制回應 — 當請求超過限額時，回傳標準 JSON 結構與 Retry-After 提示，優化前端與用戶體驗。
• 編寫 Rate Limit 自動化壓力測試腳本（test_rate_limit.py） — 模擬連續快速請求，驗證防護機制是否精準觸發。

一、 更新 requirements.txt

加入 Flask 生態系中最成熟的限流套件 Flask-Limiter：

Flask==3.0.3
google-genai==1.75.0
gTTS==2.5.1
python-dotenv==1.0.1
requests==2.32.3
Pillow
line-bot-sdk
Flask-Limiter

二、 於 app.py 整合 Rate Limiting 機制

開啟 app.py，導入 Limiter 並針對 /analyze-prescription 設置限制（例如：每分鐘最多 5 次請求）：

import os
import json
import uuid
import sqlite3
from flask import Flask, request, jsonify, send_from_directory
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from dotenv import load_dotenv
from google import genai
from google.genai import types
from PIL import Image
from gtts import gTTS

# LINE Bot SDK 引入
from linebot.v3.messaging import (
    Configuration,
    ApiClient,
    MessagingApi,
    PushMessageRequest,
    TextMessage
)

# 1. 載入環境變數與初始化
load_dotenv()
api_key = os.getenv("GEMINI_API_KEY")
line_access_token = os.getenv("LINE_CHANNEL_ACCESS_TOKEN")
line_user_id = os.getenv("LINE_USER_ID")

if not api_key:
    raise ValueError("❌ 錯誤：找不到 GEMINI_API_KEY，請檢查 .env 設定！")

client = genai.Client(api_key=api_key)

if line_access_token:
    configuration = Configuration(access_token=line_access_token)
    line_api_client = ApiClient(configuration)
    line_bot_api = MessagingApi(line_api_client)
else:
    line_bot_api = None

app = Flask(__name__)
app.json.ensure_ascii = False

# 初始化 Limiter：使用用戶 IP 作為辨識依據
limiter = Limiter(
    get_remote_address,
    app=app,
    default_limits=["200 per day", "50 per hour"],
    storage_uri="memory://"
)

# 自訂 429 Rate Limit 超限錯誤回應
@app.errorhandler(429)
def ratelimit_handler(e):
    return jsonify({
        "error": "rate_limit_exceeded",
        "message": "請求過於頻繁，系統保護中。請稍後再試。",
        "detail": str(e.description)
    }), 429

AUDIO_DIR = os.path.join(os.getcwd(), 'static', 'audio')
DATABASE_PATH = os.path.join(os.getcwd(), 'prescription_vlm.db')
os.makedirs(AUDIO_DIR, exist_ok=True)

ALLOWED_EXTENSIONS = {'png', 'jpg', 'jpeg'}

def allowed_file(filename):
    return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS

# 2. 資料庫初始化
def init_db():
    conn = sqlite3.connect(DATABASE_PATH)
    cursor = conn.cursor()

    cursor.execute('''
        CREATE TABLE IF NOT EXISTS prescriptions (
            id TEXT PRIMARY KEY,
            spoken_summary TEXT NOT NULL,
            audio_url TEXT NOT NULL,
            safety_warnings TEXT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    ''')

    cursor.execute('''
        CREATE TABLE IF NOT EXISTS medicines (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            prescription_id TEXT NOT NULL,
            name TEXT NOT NULL,
            type TEXT NOT NULL,
            frequency TEXT NOT NULL,
            dosage TEXT NOT NULL,
            timing TEXT NOT NULL,
            warning TEXT,
            FOREIGN KEY (prescription_id) REFERENCES prescriptions (id)
        )
    ''')

    conn.commit()
    conn.close()

init_db()

# 3. LINE 推播函式
def send_line_notification(summary, medicines_count, safety_warnings):
    if not line_bot_api or not line_user_id:
        print("⚠️ 尚未設定 LINE_CHANNEL_ACCESS_TOKEN 或 LINE_USER_ID，跳過推播。")
        return

    warning_text = ""
    if safety_warnings:
        warning_text = f"\n\n⚠️【用藥安全提醒】\n" + "\n".join([f"• {w}" for w in safety_warnings])

    push_text = f"💊【長輩用藥通知】\n剛才已完成藥袋辨識，共有 {medicines_count} 種藥品。{warning_text}\n\n白話摘要：\n{summary}"

    try:
        push_message_request = PushMessageRequest(
            to=line_user_id,
            messages=[TextMessage(text=push_text)]
        )
        line_bot_api.push_message(push_message_request)
        print("✅ LINE 關懷推播發送成功！")
    except Exception as e:
        print(f"❌ LINE 推播發送失敗: {str(e)}")

prescription_schema = {
    "type": "OBJECT",
    "properties": {
        "spoken_summary": {"type": "STRING"},
        "safety_warnings": {"type": "ARRAY", "items": {"type": "STRING"}},
        "medicines": {
            "type": "ARRAY",
            "items": {
                "type": "OBJECT",
                "properties": {
                    "name": {"type": "STRING"},
                    "type": {"type": "STRING"},
                    "frequency": {"type": "STRING"},
                    "dosage": {"type": "STRING"},
                    "timing": {"type": "STRING"},
                    "warning": {"type": "STRING"}
                },
                "required": ["name", "type", "frequency", "dosage", "timing"]
            }
        }
    },
    "required": ["spoken_summary", "safety_warnings", "medicines"]
}

@app.route('/ping', methods=['GET'])
def ping():
    return jsonify({"status": "online", "service": "PrescriptionVLM Engine"}), 200

# 針對 VLM 辨識 API 加入嚴格的 Rate Limit 限制（每分鐘最多 5 次）
@app.route('/analyze-prescription', methods=['POST'])
@limiter.limit("5 per minute")
def analyze_prescription():
    if 'image' not in request.files:
        return jsonify({"error": "unsupported_media_type", "message": "未提供圖片檔案"}), 400

    file = request.files['image']
    
    if file.filename == '' or not allowed_file(file.filename):
        return jsonify({
            "error": "unsupported_media_type",
            "message": "不支援的檔案格式，請上傳 .jpg, .jpeg 或 .png 圖片。"
        }), 400

    try:
        image = Image.open(file.stream)
        prompt = """
        你是一位專業且細心的藥師助手。請分析這張藥袋照片：
        1. 將藥品分類為口服或外用，精準提取名稱、頻率、劑量與吃藥時間。
        2. 檢查是否有重複藥性或高風險注意事項，填入 safety_warnings。
        3. 針對高齡長輩，撰寫一段溫柔白話的 spoken_summary。
        """

        config = types.GenerateContentConfig(
            response_mime_type="application/json",
            response_schema=prescription_schema
        )

        response = client.models.generate_content(
            model='gemini-3.6-flash',
            contents=[image, prompt],
            config=config
        )

        result_data = json.loads(response.text)

        spoken_text = result_data.get("spoken_summary", "解析完成。")
        safety_warnings = result_data.get("safety_warnings", [])

        if safety_warnings:
            prefix = "長輩請注意，這份藥單有特別需要留意的地方：" + "；".join(safety_warnings) + "。"
            spoken_text = f"{prefix} {spoken_text}"
            result_data['spoken_summary'] = spoken_text

        prescription_id = uuid.uuid4().hex[:8]
        filename = f"speech_{prescription_id}.mp3"
        filepath = os.path.join(AUDIO_DIR, filename)

        tts = gTTS(text=spoken_text, lang='zh-tw')
        tts.save(filepath)

        audio_url = f"/static/audio/{filename}"
        result_data['audio_url'] = audio_url
        result_data['prescription_id'] = prescription_id

        conn = sqlite3.connect(DATABASE_PATH)
        cursor = conn.cursor()

        warnings_json = json.dumps(safety_warnings, ensure_ascii=False)
        cursor.execute(
            "INSERT INTO prescriptions (id, spoken_summary, audio_url, safety_warnings) VALUES (?, ?, ?, ?)",
            (prescription_id, spoken_text, audio_url, warnings_json)
        )

        meds = result_data.get("medicines", [])
        for med in meds:
            cursor.execute(
                """INSERT INTO medicines 
                   (prescription_id, name, type, frequency, dosage, timing, warning) 
                   VALUES (?, ?, ?, ?, ?, ?, ?)""",
                (
                    prescription_id,
                    med.get("name"),
                    med.get("type"),
                    med.get("frequency"),
                    med.get("dosage"),
                    med.get("timing"),
                    med.get("warning", "")
                )
            )

        conn.commit()
        conn.close()

        send_line_notification(spoken_text, len(meds), safety_warnings)

        return jsonify(result_data), 200

    except Exception as e:
        return jsonify({"error": f"伺服器處理失敗: {str(e)}"}), 500

@app.route('/prescriptions', methods=['GET'])
def get_prescriptions():
    try:
        conn = sqlite3.connect(DATABASE_PATH)
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()

        cursor.execute("SELECT * FROM prescriptions ORDER BY created_at DESC")
        prescriptions = cursor.fetchall()

        history = []
        for p in prescriptions:
            cursor.execute("SELECT name, type, frequency, dosage, timing, warning FROM medicines WHERE prescription_id = ?", (p['id'],))
            meds = [dict(m) for m in cursor.fetchall()]

            history.append({
                "id": p['id'],
                "spoken_summary": p['spoken_summary'],
                "audio_url": p['audio_url'],
                "safety_warnings": json.loads(p['safety_warnings']) if p['safety_warnings'] else [],
                "created_at": p['created_at'],
                "medicines": meds
            })

        conn.close()
        return jsonify({
            "status": "success",
            "data": history
        }), 200

    except Exception as e:
        return jsonify({"error": f"查詢失敗: {str(e)}"}), 500

@app.route('/static/audio/<filename>', methods=['GET'])
def get_audio(filename):
    return send_from_directory(AUDIO_DIR, filename)

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

三、 編寫 Rate Limit 自動化測試腳本

在專案根目錄建立 test_rate_limit.py，連續發送請求驗證第 6 次是否觸發 HTTP 429：

import requests

BASE_URL = "http://127.0.0.1:5000"

def test_rate_limit():
    print("🚀 開始測試 API Rate Limit 防禦機制 (限制每分鐘 5 次)...")
    
    # 測試傳送非法檔案（觸發快速回應）連續 6 次
    files = {"image": ("test.txt", b"Hello World", "text/plain")}
    
    for i in range(1, 7):
        response = requests.post(f"{BASE_URL}/analyze-prescription", files=files)
        print(f"請求 #{i} -> Status Code: {response.status_code}")
        
        if i <= 5:
            assert response.status_code == 400
        else:
            # 第 6 次請求預期觸發 429 Too Many Requests
            assert response.status_code == 429
            data = response.json()
            assert data.get("error") == "rate_limit_exceeded"
            print("✅ [Pass] 成功觸發 HTTP 429 流量管制阻斷機制！")

if __name__ == "__main__":
    try:
        test_rate_limit()
        print("\n🎉 API Rate Limiting 防護機制驗證成功！")
    except AssertionError:
        print("\n❌ 測試失敗：限流防護未正確觸發。")
    except Exception as e:
        print(f"\n❌ 測試異常：{e}")

四、 重新打包容器與驗證測試

更新 requirements.txt 與 app.py 後，重新構建 Docker 容器並執行測試：

# 1. 強制移除舊容器並重新建構 Image（會自動安裝 Flask-Limiter）
docker rm -f prescription_service
docker build -t prescription-vlm:v1.0 .

# 2. 啟動容器
docker run -d -p 5000:5000 --env-file .env --name prescription_service prescription-vlm:v1.0

# 3. 執行 Rate Limit 測試
python test_rate_limit.py

預期輸出結果

🚀 開始測試 API Rate Limit 防禦機制 (限制每分鐘 5 次)...
請求 #1 -> Status Code: 400
請求 #2 -> Status Code: 400
請求 #3 -> Status Code: 400
請求 #4 -> Status Code: 400
請求 #5 -> Status Code: 400
請求 #6 -> Status Code: 429
✅ [Pass] 成功觸發 HTTP 429 流量管制阻斷機制！

🎉 API Rate Limiting 防護機制驗證成功！

五、 版本控制與提交 GitHub

測試成功後，將更新後的檔案提交至 GitHub：

git add .

git commit -m "保留雙引號 改填寫自己要記錄的標記 ex.鐵人賽第十二天"

git push

六、 本日小結與明日預告

今天為微服務加上了 API Rate Limiting 保護，提升了系統面對刷單與過載攻擊時的韌性。

明天（Day 13）進入 Structured Logging 與 Observability 監控，讓系統具備生產環境級別的可追蹤性與排錯能力。